package org.ws.task.schedule.dispatcher;

import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.exception.NotFoundNodeException;
import org.ws.task.item.ExecuteContext;
import org.ws.task.repository.LeaderContextRepository;
import org.ws.task.repository.NodeMetadataRepository;
import org.ws.task.model.NodeMetadata;
import org.ws.task.model.TaskItem;

import java.util.List;

public abstract class AbstractDispatcher implements Dispatcher {

    @Override
    public void dispatch(TaskItem item, String taskName) {
        List<NodeMetadata> taskWorkerNodes = NodeMetadataRepository.getInstance().getActiveWorkerNodes(taskName);
        if(CollectionUtils.isNotEmpty(taskWorkerNodes)){
            List<NodeMetadata> workerNodes = getNode(item,taskWorkerNodes);
            if(CollectionUtils.isNotEmpty(workerNodes)){
                for (NodeMetadata workerNode : workerNodes) {
                    String id = workerNode.getId();
                    ExecuteContext context = LeaderContextRepository.getInstance().getContext(taskName, id);
                    context.add(item);
                }
                return;
            }
        }
        throw new NotFoundNodeException("task["+taskName+"] can not fond worker node.");
    }

    protected abstract List<NodeMetadata> getNode(TaskItem item,List<NodeMetadata> taskWorkerNodes);
}
